// 加锁逻辑 
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) { 
    if (leaseTime != -1) { 
 return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); 
    } 
    // 调用一段lua脚本，设置一些key、过期时间 
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); 
    ttlRemainingFuture.addListener(new FutureListener<Long>() { 
        @Override 
 public void operationComplete(Future<Long> future) throws Exception { 
            if (!future.isSuccess()) { 
 return; 
            } 
 
            Long ttlRemaining = future.getNow(); 
            // lock acquired 
            if (ttlRemaining == null) { 
                // 看门狗逻辑 
                scheduleExpirationRenewal(threadId); 
            } 
        } 
    }); 
 return ttlRemainingFuture; 
} 
 
 
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { 
    internalLockLeaseTime = unit.toMillis(leaseTime); 
 
 return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command, 
 "if (redis.call('exists', KEYS[1]) == 0) then " + 
 "redis.call('hset', KEYS[1], ARGV[2], 1); " + 
 "redis.call('pexpire', KEYS[1], ARGV[1]); " + 
 "return nil; " + 
 "end; " + 
 "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + 
 "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + 
 "redis.call('pexpire', KEYS[1], ARGV[1]); " + 
 "return nil; " + 
 "end; " + 
 "return redis.call('pttl', KEYS[1]);", 
                Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); 
} 
 
 
 
// 看门狗最终会调用了这里 
private void scheduleExpirationRenewal(final long threadId) { 
    if (expirationRenewalMap.containsKey(getEntryName())) { 
 return; 
    } 
 
    // 这个任务会延迟10s执行 
    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { 
        @Override 
 public void run(Timeout timeout) throws Exception { 
 
            // 这个操作会将key的过期时间重新设置为30s 
            RFuture<Boolean> future = renewExpirationAsync(threadId); 
 
            future.addListener(new FutureListener<Boolean>() { 
                @Override 
 public void operationComplete(Future<Boolean> future) throws Exception { 
                    expirationRenewalMap.remove(getEntryName()); 
                    if (!future.isSuccess()) { 
                        log.error("Can't update lock " + getName() + " expiration", future.cause()); 
 return; 
                    } 
 
                    if (future.getNow()) { 
                        // reschedule itself 
                        // 通过递归调用本方法，无限循环延长过期时间 
                        scheduleExpirationRenewal(threadId); 
                    } 
                } 
            }); 
        } 
 
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); 
 
    if (expirationRenewalMap.putIfAbsent(getEntryName(), new ExpirationEntry(threadId, task)) != null) { 
        task.cancel(); 
    } 
} 